package TestSparkStreaming;


import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function0;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.storage.StorageLevel;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**8.2.DStream的有状态操作
 * 8.2.1.updateStateByKey
 * 该算子使得前面所有批次的数据可以和当前批次的数据累加。
 * 例如“从程序运行到现在，累计接收到的每个单词数量分别是多少”场景就可以使用该算子实现。如下图所示：
 */
public class TestSparkStreaming3 {
    public static final String checkpointDir = "checkpoint_dir/";
    public static void main(String[] args) {
        JavaStreamingContext orCreate = JavaStreamingContext.getOrCreate(checkpointDir, new MyFun0());
        orCreate.start();
        try {
            orCreate.awaitTermination();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


    }

    public static class MyFun0 implements Function0<JavaStreamingContext> {



        @Override
        public JavaStreamingContext call() throws Exception {
            // 创建 SparkConf 对象
            SparkConf sparkConf = new SparkConf().setAppName(TestSparkStreaming.class.getName())
                    // 设置master 本地运行设置 开启两个线程
                    .setMaster("local[2]");
            // 创建 sparkContext 对象
            JavaSparkContext sparkContext = new JavaSparkContext(sparkConf);
            // 基于 sparkContext 设置 StreamingContext
            JavaStreamingContext streamingContext = new JavaStreamingContext(sparkContext, new Duration(1000 * 10));

            // 将数据持久化 当前重启时 可以将前面累计的数据重新加载进来
            streamingContext.checkpoint(checkpointDir);

            // 接收 端口 传递过来的数据  参数1 监听 端口的进程所在的IP  参数2 监听的端口号 参数3 设置储存级别
            JavaReceiverInputDStream<String> dStream = streamingContext.socketTextStream("127.0.0.1", 9999, StorageLevel.MEMORY_AND_DISK());

            JavaPairDStream<String, Integer> stringIntegerJavaPairDStream = dStream.flatMap(new FlatMapFunction<String, String>() {
                @Override
                public Iterator<String> call(String s) throws Exception {
                    String[] s1 = s.split(" ");
                    return Arrays.asList(s1).iterator();
                }
            }).mapToPair(new PairFunction<String, String, Integer>() {
                @Override
                public Tuple2<String, Integer> call(String s) throws Exception {
                    return new Tuple2<>(s, 1);
                }
            }).updateStateByKey((Function2<List<Integer>, Optional<Integer>, Optional<Integer>>) (value, state) -> {
                Integer sum = 0;
                if (state.isPresent()){sum = state.get();}
                for (Integer v:value) {
                    sum += v;
                }
                return Optional.of(sum);

            });
            stringIntegerJavaPairDStream.print();


            return streamingContext;
        }
    }


}

